package own.stu.jobgib.playown.netty.teston;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import own.stu.jobgib.playown.netty.teston.httpClient.RapidConfig;
import own.stu.jobgib.playown.netty.teston.queue.NettyBatchEventProcessor;
import own.stu.jobgib.playown.netty.teston.queue.NettyCoreProcessor;
import own.stu.jobgib.playown.netty.teston.queue.NettyProcessor;
import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.ThreadPerTaskExecutor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class Server {

    public static final boolean OPEN = false;

    //	http body报文最大大小
    private static int maxContentLength = 8 * 1024 * 1024;
    private static int port = 8888;

    public static void main(String[] args) throws InterruptedException {

        RapidConfig rapidConfig = new RapidConfig();
        NettyCoreProcessor nettyCoreProcessor = new NettyCoreProcessor();
        NettyProcessor nettyProcessor = new NettyBatchEventProcessor(rapidConfig, nettyCoreProcessor);

        EventLoopGroup boss = new NioEventLoopGroup(1, new ThreadPerTaskExecutor(new DefaultThreadFactory("boss")));
        // EventLoopGroup work = new NioEventLoopGroup(new DefaultThreadFactory("work"));
        EventLoopGroup work = new NioEventLoopGroup(3, new DefaultThreadFactory("work"));

        ServerBootstrap bootstrap = new ServerBootstrap();
        ServerBootstrap handler = bootstrap.group(boss, work)
                .channel(NioServerSocketChannel.class)

                // 临时存放已完成三次握手的请求的队列的最大长度。
                // 如果未设置或所设置的值小于1，Java将使用默认值50。
                // 如果大于队列的最大长度，请求会被拒绝
                .option(ChannelOption.SO_BACKLOG, 4)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, 65535)
                .childOption(ChannelOption.SO_RCVBUF, 65535)
                .localAddress(port)
                .childHandler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline().addLast(new HttpServerCodec(),
                                new HttpObjectAggregator(maxContentLength),
                                new HttpServerExpectContinueHandler(),
                                new NettyServerConnectManagerHandler(),
                                new NettyHttpServerHandler(nettyProcessor));
                    }
                });

        handler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

        bootstrap.bind().sync();
        log.info("< ============= Server StartUp On Port: " + port + "================ >");

        nettyProcessor.start();
        log.info("< ============= flusher queue start up ================ >");

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            if (boss != null) {
                boss.shutdownGracefully();
            }
            if (work != null) {
                work.shutdownGracefully();
            }
            if (nettyProcessor != null) {
                nettyProcessor.shutdown();
            }
        }, "server-hook"));
    }

    static class NettyHttpServerHandler extends ChannelInboundHandlerAdapter {

        private NettyProcessor<own.stu.jobgib.playown.netty.teston.HttpRequestWrapper> nettyProcessor;

        public NettyHttpServerHandler(NettyProcessor<own.stu.jobgib.playown.netty.teston.HttpRequestWrapper> nettyProcessor) {
            this.nettyProcessor = nettyProcessor;
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof HttpRequest) {

                FullHttpRequest request = (FullHttpRequest) msg;
                own.stu.jobgib.playown.netty.teston.HttpRequestWrapper httpRequestWrapper = new own.stu.jobgib.playown.netty.teston.HttpRequestWrapper();
                httpRequestWrapper.setFullHttpRequest(request);
                httpRequestWrapper.setCtx(ctx);

                nettyProcessor.process(httpRequestWrapper);
            }
        }
    }

    static class NettyServerConnectManagerHandler extends ChannelDuplexHandler {

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            String remoteAddr = own.stu.jobgib.playown.netty.teston.RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            if (log.isInfoEnabled() && OPEN) {
                log.info("NETTY SERVER PIPLINE: channelRegistered {}", remoteAddr);
            }
            super.channelRegistered(ctx);
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            String remoteAddr = own.stu.jobgib.playown.netty.teston.RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            if (log.isInfoEnabled() && OPEN) {
                log.info("NETTY SERVER PIPLINE: channelUnregistered {}", remoteAddr);
            }
            super.channelUnregistered(ctx);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            String remoteAddr = own.stu.jobgib.playown.netty.teston.RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            if (log.isInfoEnabled() && OPEN) {
                log.info("NETTY SERVER PIPLINE: channelActive {}", remoteAddr);
            }
            super.channelActive(ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            String remoteAddr = own.stu.jobgib.playown.netty.teston.RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            if (log.isInfoEnabled() && OPEN) {
                log.info("NETTY SERVER PIPLINE: channelInactive {}", remoteAddr);
            }
            super.channelInactive(ctx);
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
                if (idleStateEvent.state().equals(IdleState.ALL_IDLE)) {
                    String remoteAddr = own.stu.jobgib.playown.netty.teston.RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                    log.warn("NETTY SERVER PIPLINE: userEventTriggered: IDLE {}", remoteAddr);
                    ctx.channel().close();
                }
            }
            super.userEventTriggered(ctx, evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            String remoteAddr = own.stu.jobgib.playown.netty.teston.RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.warn("NETTY SERVER PIPLINE: remoteAddr: {}, exceptionCaught: ", remoteAddr, cause);
            ctx.channel().close();
        }
    }
}
